package com.lry.mqstarter.extend;

import com.alibaba.fastjson.JSON;
import com.lry.mqstarter.properties.MqConsumerProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import com.lry.mqstarter.annotation.RocketMQListener;
import com.lry.mqstarter.build.ConsumerBuild;
import com.lry.mqstarter.properties.MqConsumerDetailProperties;

import java.util.Map;
import java.util.stream.Collectors;


@Slf4j
public class MqConsumerSmartInitializingSingleton implements SmartInitializingSingleton, ApplicationContextAware {

    private ApplicationContext applicationContext;

    private MqConsumerProperties mqConsumerProperties;


    public MqConsumerSmartInitializingSingleton(MqConsumerProperties mqConsumerProperties) {
        this.mqConsumerProperties = mqConsumerProperties;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void afterSingletonsInstantiated() {

        startSelfMqConsumer();

    }


    private void startSelfMqConsumer() {

        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQListener.class)
                .entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        log.info("初始化mq消费者配置-defaultProp--------{}", JSON.toJSONString(mqConsumerProperties.getDefaultProp()));

        beans.forEach(this::registerSelfContainer);
    }


    private void registerSelfContainer(String beanName, Object bean) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);

        MqConsumerDetailProperties defaultProp = mqConsumerProperties.getDefaultProp();

        RocketMQListener annotation = clazz.getAnnotation(RocketMQListener.class);
        MqConsumerDetailProperties properties = new MqConsumerDetailProperties();
        properties.setBeanName(beanName);
        properties.setConsumerGroup(annotation.consumerGroup());
        properties.setConsumeMessageBatchMaxSize(annotation.consumeMessageBatchMaxSize());
        properties.setInstanceName(annotation.instanceName());
        properties.setTopic(annotation.topic());
        properties.setNameSrvAddr(annotation.nameSrvAddr());
        properties.setSubExpression(annotation.subExpression());
        properties.setMessageModel(annotation.messageModel());
        ConsumerBuild.buildSelfMQConsumer(bean,properties, defaultProp);
    }

}
